package com.example.klinemqflink.service;

import com.example.klinemqflink.model.MarketData;
import com.example.klinemqflink.model.RMDeserializationSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.rabbitmq.RMQSource;
import org.apache.flink.streaming.connectors.rabbitmq.common.RMQConnectionConfig;
import org.springframework.stereotype.Service;

@Service
public class FlinkRabbitMQTopicConsumer {
    private static final String EXCHANGE_NAME = "topic_trades";
    private static final String QUEUE_NAME = "flink_topic_queue";
    private static final String ROUTING_KEY_PATTERN = "stocks.*";

    public static void main(String[] args) throws Exception {
        // 创建 Flink 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 配置 RabbitMQ 连接
        RMQConnectionConfig connectionConfig = new RMQConnectionConfig.Builder()
                .setHost("localhost")
                .setPort(5672)
                .setUserName("guest")
                .setPassword("guest")
                .setVirtualHost("/")
                .build();

        // 声明队列并绑定到 topic 交换机
        try (com.rabbitmq.client.Connection connection = new com.rabbitmq.client.ConnectionFactory().newConnection();
             com.rabbitmq.client.Channel channel = connection.createChannel()) {
            channel.exchangeDeclare(EXCHANGE_NAME, "topic");
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY_PATTERN);
        }

        // 创建 RMQSource
        RMQSource<MarketData> source = new RMQSource<>(
                connectionConfig,
                QUEUE_NAME,
                true,
                new RMDeserializationSchema()
        );

        // 创建 DataStream
        DataStream<MarketData> stream = env.addSource(source);

        // 处理数据流，这里简单打印
        stream.print();

        // 执行 Flink 作业
        env.execute("Flink RabbitMQ Topic Consumer");
    }
}
